package com.goomba.transation;


import com.goomba.demo.contants.Commons;
import io.netty.util.internal.StringUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.producer.*;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;

import java.nio.charset.StandardCharsets;
import java.sql.ResultSet;

/**
 * 通过事务提交数据
 * 整个流程上，我们先执行半消息。就和我们之前发送是一样的。半消息成功之后。MQ会回调我们的本地事务状态。如果是commit/rollback就执行
 * 如果是unknow 那么就执行checkLocalTransaction 进行检查。检查之后在进行判断是否提交事务
 * @author goomba
 */
public class Producer {

    public static void main(String[] args) throws Exception {
        TransactionMQProducer producer = new TransactionMQProducer("transaction_group");

        producer.setNamesrvAddr(Commons.nameSrvAddr);


        // 事务接收器
        producer.setTransactionListener(new TransactionListener() {

            /**
             * 执行本地事务。
             * @param message
             * @param o
             * @return
             */
            @Override
            public LocalTransactionState executeLocalTransaction(Message message, Object o) {


                if (StringUtils.equals("TAGA", message.getTags())) {
                    return LocalTransactionState.COMMIT_MESSAGE;

                } else if (StringUtils.equals("TAGB", message.getTags())) {
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                }
                System.out.println("TAG IS " + message.getTags());
                return LocalTransactionState.UNKNOW;
            }


            /**
             * 事务如果没有commit/rollback, MQ会进行回查事务
             * @param messageExt
             * @return
             */
            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {

                System.out.println("invoke checkLocalTransaction tags is " + messageExt.getTags());

                return LocalTransactionState.COMMIT_MESSAGE;
            }
        });

        producer.start();
        String[] tags = {"TAGA", "TAGB", "TAGC"};

        for (int i = 0; i < 3; i++) {
            Message message = new Message("TransactionTopic", tags[i], ("Transaction_" + i).getBytes(StandardCharsets.UTF_8));

            SendResult result = producer.sendMessageInTransaction(message, null);

            System.out.println("返回的结果是：" +  result.getSendStatus());

        }

//        producer.shutdown();

    }
}
